package com.atbeijing.D06;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;


// 触发器
public class Example8 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .socketTextStream("localhost", 9999)
                .map(new MapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String value) throws Exception {
                        String[] arr = value.split(" ");
                        return Tuple2.of(arr[0], Long.parseLong(arr[1]) * 1000L);
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                                        return element.f1;
                                    }
                                })
                )
                .keyBy(r -> r.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .trigger(new MyTrigger())
                .process(new ProcessWindowFunction<Tuple2<String, Long>, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<Tuple2<String, Long>> elements, Collector<String> out) throws Exception {
                        out.collect(context.window().getStart() + "~" + context.window().getEnd() + "的窗口触发计算了，共 " + elements.spliterator().getExactSizeIfKnown() + " 条元素");
                    }
                })
                .print();

        env.execute();
    }

    public static class MyTrigger extends Trigger<Tuple2<String, Long>, TimeWindow> {
        @Override
        public TriggerResult onElement(Tuple2<String, Long> element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
            // 每来一条数据调用一次
            ValueState<Boolean> firstUpdated = ctx.getPartitionedState(new ValueStateDescriptor<Boolean>("first-updated", Types.BOOLEAN));
            // 当第一条数据到达的时候
            if (firstUpdated.value() == null) {
                // 当前水位线接下来的整数秒
                long ts = ctx.getCurrentWatermark() + (1000L - ctx.getCurrentWatermark() % 1000L);
                System.out.println("窗口" + window.getStart() + "~" + window.getEnd() + "的第一个元素到来了，" + "水位线是：" + ctx.getCurrentWatermark());
                System.out.println("窗口" + window.getStart() + "~" + window.getEnd() + "的第一个元素到来了，" + "注册的定时器时间戳是：" + ts);
                ctx.registerEventTimeTimer(ts); // 注册onEventTime定时器
                System.out.println("窗口" + window.getStart() + "~" + window.getEnd() + "的第一个元素到来了，" + "注册的定时器时间戳是窗口结束时间，是：" + window.getEnd());
                ctx.registerEventTimeTimer(window.getEnd()); // 注册窗口结束时间的定时器
                firstUpdated.update(true);
            }
            return TriggerResult.CONTINUE;
        }

        @Override
        public TriggerResult onProcessingTime(long timerTs, TimeWindow window, TriggerContext ctx) throws Exception {
            //定时器设置的是处理时间时  机器时间到达timerTs时，触发定时器onProcessingTime函数的执行
            return TriggerResult.CONTINUE;
        }

        @Override
        public TriggerResult onEventTime(long timerTs, TimeWindow window, TriggerContext ctx) throws Exception {
            //定时器设置的是事件时间时 水位线到达timerTs时，触发定时器onEventTime函数的执行
            if (timerTs == window.getEnd()) {
                // 是`ctx.registerEventTimeTimer(window.getEnd());`这一句注册的定时器
                return TriggerResult.FIRE_AND_PURGE;
            } else {
                System.out.println("窗口" + window.getStart() + "~" + window.getEnd() + "的onEventTime里的水位线的值是：" + ctx.getCurrentWatermark());
                long ts = ctx.getCurrentWatermark() + (1000L - ctx.getCurrentWatermark() % 1000L);
                if (ts < window.getEnd()) {
                    System.out.println("窗口" + window.getStart() + "~" + window.getEnd() + "在onEventTime中注册了定时器，时间戳是：" + ts);
                    ctx.registerEventTimeTimer(ts);
                }
                return TriggerResult.FIRE;
            }
        }

        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
            // 窗口销毁时调用
            // 单例，只会被实例化一次
            ValueState<Boolean> firstUpdated = ctx.getPartitionedState(new ValueStateDescriptor<Boolean>("first-updated", Types.BOOLEAN));
            firstUpdated.clear();
        }
    }
}
